package kafkastudy.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Created by hp on 2018/8/6.
 */
public class ActivemqConsumer {

    public static void main(String[] args) {

        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        try {
            //1创建连接工厂ConnectionFactory  需要username/password/url
            System.out.println(ActiveMQConnectionFactory.DEFAULT_USER + ActiveMQConnectionFactory.DEFAULT_PASSWORD
                    + ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
            connectionFactory = new ActiveMQConnectionFactory("admin",
                    "admin", "failover:(tcp://10.1.1.49:61617,tcp://10.1.1.75:61617,tcp://10.1.1.80:61617)?Randomize=false");

            //2创建连接Connection 注意连接默认是关闭的，因此需要start开启
            connection = connectionFactory.createConnection();
            connection.start();

            //3前面2步的操作其实就是为了创建Session（上下文环境对象）
            //创建session有一些配置参数 比如是否启用事务，签收模式（这里先设置为自动签收模式auto）
            session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //4通过session创建Destination对象，在p2p模式下是queue，在pub/sub模式下是topic
            Destination destination = session.createQueue("queue1");

            //5通过session， 发送消息的生产者，接收消息的消费者
            consumer = session.createConsumer(destination);

            //6消费者同步接收
            Message message = consumer.receive(1000);
            System.out.println("Received message: " + message);

            while(true) {
                //7消费者异步接收
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        System.out.println("Received message: " + message);
                    }
                });
            }


        } catch (JMSException e) {
            e.printStackTrace();
        } finally {
            try {
                consumer.close();
                session.close();
                connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}
